package online.duanqing.mqspringboot.service;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;

/**
 * Created with FR
 *
 * @Author: DQ
 * @Date: 2024/10/18/9:55
 * @Description: 描述语句
 */
@Component
@RocketMQMessageListener(messageModel = MessageModel.CLUSTERING, consumerGroup = "Pro_Group", topic = "topic-A")
@Slf4j
public class MQConsumerService implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(String s) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {


        log.info("----------------------主题{},开始监听消费----------------------","topic-A");

        defaultMQPushConsumer.setPullInterval(1000 * 2);//设置拉取消息间隔，单位毫秒
        defaultMQPushConsumer.setConsumeThreadMin(2);//最小消费线程数
        defaultMQPushConsumer.setConsumeThreadMax(10);//最大消费线程数
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(300);//消费者单次批量消费的消息数目上限，即List<MessageExt> 的最大size
        defaultMQPushConsumer.setPullBatchSize(200);//每个队列每次拉取的最大消费数


        defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            ArrayList<JSONObject> dataList = new ArrayList<>();


            for (MessageExt messageExt : list) {
                String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);

                System.out.println(body);
            }
            log.info("本次拉取数据量为{}", list.size());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
    }
}
